package com.tempjob.admin.service.sse;

import com.tempjob.common.businesss.service.UserSessionService;
import com.tempjob.common.context.AdminContext;
import com.tempjob.common.exception.BusinessException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * 代码千万行，注释第一行，
 * 注释不规范，同事泪两行。
 * <p>
 * SseEmitter服务
 *
 * @author YuLc
 * @version 1.0.0
 * @date 2020/1/7
 */
@Service
public class SseEmitterService {

    private static final Logger logger = LoggerFactory.getLogger(SseEmitterService.class);

    @Autowired
    private UserSessionService userSessionService;

    /**
     * 当前连接数
     */
    private static AtomicInteger count = new AtomicInteger(0);

    /**
     * 使用map对象，便于根据userId来获取对应的SseEmitter，或者放redis里面
     */
    private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    /**
     * 创建用户连接并返回 SseEmitter
     *
     * @param userId 用户ID
     * @return SseEmitter
     */
    public SseEmitter connect(String userId) {
        AdminContext adminContext = userSessionService.validateTicket(userId);
        if (adminContext == null) {
            throw new BusinessException("用户登录失效!");
        }
        // 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
        SseEmitter sseEmitter = new SseEmitterUTF8(0L);
        // 注册回调
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        sseEmitter.onTimeout(timeoutCallBack(userId));
        sseEmitterMap.put(userId, sseEmitter);
        // 数量+1
        count.getAndIncrement();
        try {
//            BigScreenVO info = bigScreenService.statics(adminContext);
            sseEmitter.send("test", MediaType.APPLICATION_JSON);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        logger.info("创建新的sse连接，当前用户：{}", userId);
        return sseEmitter;
    }

    /**
     * 给指定用户发送信息
     */
    public void sendMessage(String userId) {
        if (sseEmitterMap.containsKey(userId)) {
            AdminContext adminContext = userSessionService.validateTicket(userId);
            if (adminContext != null) {
                SseEmitter sse = sseEmitterMap.get(userId);
                try {
//                    BigScreenVO info = bigScreenService.statics(adminContext);
                    sse.send(new Object(), MediaType.APPLICATION_JSON);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    /**
     * 群发消息
     */
    public void batchSendMessage(List<String> ids) {
        ids.forEach(userId -> sendMessage(userId));
    }

    /**
     * 群发所有人
     */
    public void batchSendMessage() {
        for (String k : sseEmitterMap.keySet()) {
            AdminContext adminContext = userSessionService.validateTicket(k);
            if (adminContext != null) {
                SseEmitter sse = sseEmitterMap.get(k);
                try {
//                    BigScreenVO info = bigScreenService.statics(adminContext);
                    sse.send("test", MediaType.APPLICATION_JSON);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }else{
                removeUser(k);
            }
        }
    }

    /**
     * 移除用户连接
     */
    public static void removeUser(String userId) {
        sseEmitterMap.remove(userId);
        // 数量-1
        count.getAndDecrement();
        logger.info("移除用户：{}", userId);
    }

    /**
     * 获取当前连接信息
     */
    public static List<String> getIds() {
        return new ArrayList<>(sseEmitterMap.keySet());
    }

    /**
     * 获取当前连接数量
     */
    public static int getUserCount() {
        return count.intValue();
    }

    private static Runnable completionCallBack(String userId) {
        return () -> {
            logger.info("结束连接：{}", userId);
            removeUser(userId);
        };
    }

    private static Runnable timeoutCallBack(String userId) {
        return () -> {
            logger.info("连接超时：{}", userId);
            removeUser(userId);
        };
    }

    private static Consumer<Throwable> errorCallBack(String userId) {
        return throwable -> {
            logger.info("连接异常：{}", userId);
            removeUser(userId);
        };
    }

}
